package org.xq.softcup.mq.remoting.net.common;


import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xq.softcup.mq.remoting.invoker.XQRpcInvokerFactory;
import org.xq.softcup.mq.remoting.invoker.reference.XQRpcReferenceBean;
import org.xq.softcup.mq.remoting.net.params.BaseCallback;
import org.xq.softcup.mq.remoting.net.params.XQRpcRequest;
import org.xq.softcup.mq.serialize.Serializer;

import java.util.concurrent.ConcurrentHashMap;
@Slf4j
public abstract class ConnectClient {
    private static volatile ConcurrentHashMap<String, ConnectClient> connectClientMap;

    public abstract void init(String param, Serializer serializer, XQRpcInvokerFactory rpcInvokerFactory)
            throws Exception;

    public abstract void close();

    public abstract boolean isValidate();

    public abstract void send(XQRpcRequest xqRpcRequest) throws Exception;

    public static void asyncSend(XQRpcRequest xqRpcRequest, String address,
                                 Class<? extends ConnectClient> connectClientImpl,
                                 XQRpcReferenceBean xqRpcReferenceBean) throws Exception {
        ConnectClient clientPool = getPool(address, connectClientImpl, xqRpcReferenceBean);
        try {
            clientPool.send(xqRpcRequest);
        } catch (Exception e) {
            throw e;
        }
    }

    private static volatile ConcurrentHashMap<String, Object> connectClientLockMap = new ConcurrentHashMap();

    private static ConnectClient getPool(String address, Class<? extends ConnectClient> connectClientImpl, XQRpcReferenceBean xqRpcReferenceBean)
            throws Exception {
        if (connectClientMap == null) {
            synchronized (ConnectClient.class) {
                if (connectClientMap == null) {
                    connectClientMap = new ConcurrentHashMap();

                    xqRpcReferenceBean.getInvokerFactory().addStopCallBack(new BaseCallback() {
                        public void run()
                                throws Exception {
                            if (ConnectClient.connectClientMap.size() > 0) {
                                for (String key : ConnectClient.connectClientMap.keySet()) {
                                    ConnectClient clientPool = (ConnectClient) ConnectClient.connectClientMap.get(key);
                                    clientPool.close();
                                }
                                ConnectClient.connectClientMap.clear();
                            }
                        }
                    });
                }
            }
        }
        ConnectClient connectClient = (ConnectClient) connectClientMap.get(address);
        if ((connectClient != null) && (connectClient.isValidate())) {
            return connectClient;
        }
        Object clientLock = connectClientLockMap.get(address);
        if (clientLock == null) {
            connectClientLockMap.putIfAbsent(address, new Object());
            clientLock = connectClientLockMap.get(address);
        }
        synchronized (clientLock) {
            connectClient = (ConnectClient) connectClientMap.get(address);
            if ((connectClient != null) && (connectClient.isValidate())) {
                return connectClient;
            }
            if (connectClient != null) {
                connectClient.close();
                connectClientMap.remove(address);
            }
            ConnectClient connectClient_new = (ConnectClient) connectClientImpl.newInstance();
            connectClient_new.init(address, xqRpcReferenceBean.getSerializer(), xqRpcReferenceBean.getInvokerFactory());
            connectClientMap.put(address, connectClient_new);

            return connectClient_new;
        }
    }
}
